package cn.doitedu.flink.demos;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class DynamicRules {
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> rule = env.addSource(SourcesTest.createSocketSource("localhost", 9988)).setParallelism(1);

        KeyedStream<String, String> source = env.addSource(SourcesTest.createSocketSource("localhost", 9999)).keyBy(new KeySelector<String, String>() {

            @Override
            public String getKey(String s) throws Exception {
                String uid = s.split(",")[0];
                return uid;
            }
        });

        MapStateDescriptor<String, String> ruleDesc = new MapStateDescriptor<>("rule_desc", String.class, String.class);
        BroadcastStream<String> broadcast = rule.broadcast(ruleDesc);

        BroadcastConnectedStream<String, String> connect = source.connect(broadcast);
        connect.process(new KeyedBroadcastProcessFunction<String, String, String, String>() {

            Logger logger = LoggerFactory.getLogger(DynamicRules.class.getName());
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);

            }

            @Override
            public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                String[] fields = value.split(",");
                String uid = fields[0];
                String ruleId = fields[1];
                String rule = ctx.getBroadcastState(ruleDesc).get(ruleId);
                System.out.println("收到一条事件数据: " + value);
                System.out.println("事件的uid为: "+ uid);
                System.out.println("事件的匹配rule id: " + ruleId);
                System.out.println("事件的匹配rule内容为: " + rule);
            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                String[] ruleFields = value.split(",");
                BroadcastState<String, String> bs = ctx.getBroadcastState(ruleDesc);
                bs.put(ruleFields[0],ruleFields[1]);

                System.out.println("收到一条rule，并放入了state，当前state中的数据为：");
                for (Map.Entry<String, String> entry : bs.entries()) {
                    System.out.println(entry.getKey()+" | " + entry.getValue());
                }

            }
        }).print();



        env.execute();

    }
}
